home *** CD-ROM | disk | FTP | other *** search
/ MACD 5 / MACD 5.bin / workbench / libs / queue.lzh / queue_3.1 / queue_library.c < prev    next >
Encoding:
C/C++ Source or Header  |  1996-12-02  |  10.7 KB  |  478 lines
  1. /*
  2.    queue_library.c --- queue library interface.
  3.  
  4.    Copyright (c) 1995 SHW Wabnitz
  5.    Written by Bernhard Fastenrath (fasten@shw.com)
  6.  
  7.    This file may be distributed under the terms
  8.    of the GNU General Public License.
  9. */
  10.  
  11. #if defined (__GNUC__)
  12. #include <stabs.h>
  13. #endif
  14.  
  15. #include "queue_library.h"
  16.  
  17. struct ExecBase *SysBase = NULL;
  18. struct Library *QueueBase = NULL;
  19. Semaphore QueuesSemaphore;
  20. List Queues;
  21.  
  22. #if defined (__GNUC__)
  23. const BYTE LibName[] = "queue.library";
  24. const BYTE LibIdString[] = "$VER: queue.library 3.1 (12-3-96)";
  25. const UWORD LibVersion = 3;
  26. const UWORD LibRevision = 1;
  27. #endif
  28.  
  29. #if defined (__GNUC__)
  30. #define LIBRT
  31. #define REG(regname)
  32. #define STRUCT_MYLIB struct Library
  33. #elif defined (__SASC)
  34. #define LIBRT __saveds __asm
  35. #define REG(regname) register __ ## regname
  36. #define ADDTABL_1(name,arg1);
  37. #define ADDTABL_2(name,arg1,arg2);
  38. #define ADDTABL_3(name,arg1,arg2,arg3);
  39. #define ADDTABL_END();
  40. #define STRUCT_MYLIB struct MyLibrary
  41. #endif
  42.  
  43. /*** configuration ***/
  44.  
  45. #ifdef SERIALIZE_WITH_FORBID
  46. #define ObtainSemaphore(s)  Forbid()
  47. #define ReleaseSemaphore(s) Permit()
  48. #endif
  49.  
  50. /*** internal functions ***/
  51.  
  52. #if defined (__GNUC__)
  53. static ULONG
  54. strlen (char *name)
  55. {
  56.   ULONG t = 0;
  57.  
  58.   while (name[t])
  59.     t++;
  60.   return t;
  61. }
  62. #endif
  63.  
  64. static void
  65. SetMarker (QueueHandle *qh, QMessage *msg)
  66. {
  67.   QueueNode *qn = qh -> qh_QNode;
  68.  
  69.   if (qh -> qh_un.qhl.qhl_MinNode.mln_Succ)
  70.     Remove ((Node *) &qh -> qh_un.qhl.qhl_MinNode);
  71.   Insert (&qn -> qn_List, (Node *) &qh -> qh_un.qhl.qhl_MinNode, (Node *) msg);
  72. }
  73.  
  74. static void
  75. ClearMarker (QueueHandle *qh)
  76. {
  77.   Remove ((Node *) &qh -> qh_un.qhl.qhl_MinNode);
  78.   qh -> qh_un.qhl.qhl_MinNode.mln_Succ = NULL;
  79. }
  80.  
  81. static void
  82. RemoveAndReply (QMessage *msg)
  83. {
  84.   QueueHandle *qh;
  85.  
  86.   if (!msg -> qm_Refs)
  87.   {
  88.     qh = (QueueHandle *) msg -> qm_Owner;
  89.     Remove ((Node *) msg);
  90.     /*** reply to owner ***/
  91.     /* ObtainSemaphore (); */
  92.     AddHead (&qh -> qh_un.qhs.qhs_ReplyList, (Node *) msg);
  93.     Signal (qh -> qh_SigTask, qh -> qh_SigMask);
  94.     /* ReleaseSemaphore (); */
  95.   }
  96.   else
  97.     msg -> qm_Status = QMS_REMOVED;
  98. }
  99.  
  100. static void
  101. ReplyQMessage (QueueNode *qn, QMessage *msg)
  102. {
  103.   msg -> qm_Refs --;
  104.   msg -> qm_Replies ++;
  105.  
  106.   if (msg -> qm_Replies >= qn -> qn_Read || msg -> qm_Status == QMS_REMOVED)
  107.     RemoveAndReply (msg);
  108. }
  109.  
  110. /*** library interface ***/
  111.  
  112. int LIBRT
  113. __UserLibInit (REG(a6) STRUCT_MYLIB *libbase)
  114. {
  115.   SysBase = *(struct ExecBase **)4;
  116.   QueueBase = (struct Library *) libbase;
  117.  
  118. #ifndef SERIALIZE_WITH_FORBID
  119.   InitSemaphore (&QueuesSemaphore);
  120. #endif
  121.   NewList (&Queues);
  122.   return 0; /* success */
  123. }
  124.  
  125. void LIBRT
  126. __UserLibCleanup (REG(a6) STRUCT_MYLIB *libbase)
  127. {
  128. }
  129.  
  130. ADDTABL_3(LIBQOpen,a0,d0,d1);
  131.  
  132. QHandle LIBRT
  133. LIBQOpen (REG(a0) STRPTR name, REG(d0) ULONG mode, REG(d1) ULONG sigbit)
  134. {
  135.   QueueHandle *qh;
  136.   QueueNode *qn;
  137.   ULONG len;
  138.  
  139.   if (!(qh = (QueueHandle *) AllocMem (sizeof (QueueHandle), MEMF_PUBLIC | MEMF_CLEAR)))
  140.     return NULL;
  141.  
  142.   ObtainSemaphore (&QueuesSemaphore);
  143.  
  144.   if (!(qn = (QueueNode *) FindName (&Queues, name)))
  145.   {
  146.     if (!(qn = AllocMem (sizeof (QueueNode), MEMF_PUBLIC | MEMF_CLEAR)))
  147.     {
  148.       ReleaseSemaphore (&QueuesSemaphore);
  149.       FreeMem (qh, sizeof (QueueHandle));
  150.       return NULL;
  151.     }
  152. #ifndef SERIALIZE_WITH_FORBID
  153.     InitSemaphore (&qn -> qn_Semaphore);
  154. #endif
  155.     ObtainSemaphore (&qn -> qn_Semaphore);
  156.  
  157.     len = strlen (name) + 1;
  158.     if (!(qn -> qn_Node.ln_Name = AllocMem (len, MEMF_PUBLIC)))
  159.     {
  160.       ReleaseSemaphore (&qn -> qn_Semaphore);
  161.       ReleaseSemaphore (&QueuesSemaphore);
  162.       FreeMem (qh, sizeof (QueueHandle));
  163.       FreeMem (qn, sizeof (QueueNode));
  164.       return NULL;
  165.     }
  166.     CopyMem (name, qn -> qn_Node.ln_Name, len);
  167.  
  168.     AddHead (&Queues, (Node *) qn);
  169.     NewList (&qn -> qn_Handles);
  170.     NewList (&qn -> qn_List);
  171.     qn -> qn_Refs = 1;
  172.   }
  173.   else
  174.   {
  175.     ObtainSemaphore (&qn -> qn_Semaphore);
  176.     qn -> qn_Refs ++;
  177.   }
  178.   AddHead (&qn -> qn_Handles, (Node *) qh);
  179.   qh -> qh_Mode    = mode;
  180.   qh -> qh_QNode   = qn;
  181.   qh -> qh_SigMask = 1 << sigbit;
  182.   qh -> qh_SigTask = FindTask (0);
  183.   if (mode == QMODE_LISTEN)
  184.   {
  185.     qn -> qn_Read ++;
  186.     qh -> qh_un.qhl.qhl_Status = QMS_MARKER;
  187.     Signal (qh -> qh_SigTask, qh -> qh_SigMask);
  188.   }
  189.   else
  190.   {
  191.     NewList (&qh -> qh_un.qhs.qhs_ReplyList);
  192.   }
  193.   ReleaseSemaphore (&qn -> qn_Semaphore);
  194.   ReleaseSemaphore (&QueuesSemaphore);
  195.   return (QHandle) qh;
  196. }
  197.  
  198. ADDTABL_1(LIBQClose,a0);
  199.  
  200. ULONG LIBRT
  201. LIBQClose (REG(a0) QHandle qhandle)
  202. {
  203.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  204.   QueueHandle *qh = qhandle;
  205.   QMessage *cmsg, *msg, *next;
  206.   ULONG count;
  207.  
  208.   /* to avoid a possible deadlock with QOpen() the QueuesSemaphore
  209.      has to be locked before the QNode's semaphore here.
  210.   */
  211.   ObtainSemaphore (&QueuesSemaphore);
  212.   ObtainSemaphore (&qn -> qn_Semaphore);
  213.  
  214.   if (qh -> qh_Mode == QMODE_SEND)
  215.   {
  216.     if (count = qh -> qh_un.qhs.qhs_MsgCount)
  217.     {
  218.       ReleaseSemaphore (&QueuesSemaphore);
  219.       ReleaseSemaphore (&qn -> qn_Semaphore);
  220.       return count;
  221.     }
  222.   }
  223.   else
  224.   {
  225.     if (cmsg = (QMessage *) qh -> qh_un.qhl.qhl_MinNode.mln_Succ)
  226.     {
  227.       ClearMarker (qh);
  228.       next = (QMessage *) cmsg -> qm_MinNode.mln_Pred;
  229.     }
  230.     else if (cmsg = qh -> qh_un.qhl.qhl_Message)
  231.     {
  232.       cmsg -> qm_Refs --;
  233.       cmsg -> qm_Replies ++;
  234.       next = cmsg;
  235.     }
  236.     if (next)
  237.     {
  238.       for (msg = next; next = (QMessage *) msg -> qm_MinNode.mln_Pred;
  239.        msg = next)
  240.       {
  241.     /* Replies from the current task
  242.        don't count after QClose().
  243.         */
  244.     if (msg -> qm_Status == QMS_ACTIVE || msg -> qm_Status == QMS_REMOVED)
  245.       msg -> qm_Replies --;
  246.       }
  247.     }
  248.     else
  249.       cmsg = (QMessage *) qn -> qn_List.lh_Head;
  250.  
  251.     for (msg = cmsg; next = (QMessage *) msg -> qm_MinNode.mln_Succ; msg = next)
  252.     {
  253.       /* These messages might be in the queue, waiting only
  254.      for the current task; let's give them a chance.
  255.       */
  256.       if ((msg -> qm_Status == QMS_ACTIVE && msg -> qm_Replies >= qn -> qn_Read)
  257.     || msg -> qm_Status == QMS_REMOVED)
  258.       {
  259.         RemoveAndReply (msg);
  260.       }
  261.       if (msg -> qm_Status == QMS_MARKER)
  262.     break;
  263.     }
  264.     qn -> qn_Read --;
  265.   }
  266.   Remove ((Node *) qh);
  267.   FreeMem (qh, sizeof (QueueHandle));
  268.  
  269.   if (! -- qn -> qn_Refs)
  270.   {
  271.     ReleaseSemaphore (&qn -> qn_Semaphore);
  272.     Remove ((Node *) qn);
  273.     FreeMem (qn -> qn_Node.ln_Name, strlen (qn -> qn_Node.ln_Name) + 1);
  274.     FreeMem (qn, sizeof (QueueNode));
  275.   }
  276.   else
  277.     ReleaseSemaphore (&qn -> qn_Semaphore);
  278.  
  279.   ReleaseSemaphore (&QueuesSemaphore);
  280.   return 0;
  281. }
  282.  
  283. ADDTABL_2(LIBQAddMsg,a0,a1);
  284.  
  285. void LIBRT
  286. LIBQAddMsg (REG(a0) QHandle qhandle, REG(a1) QMessage *msg)
  287. {
  288.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  289.   QueueHandle *qh = (QueueHandle *) qhandle;
  290.  
  291.   msg -> qm_Owner   = qhandle;
  292.   msg -> qm_Refs    = 0;
  293.   msg -> qm_Replies = 0;
  294.   msg -> qm_Status  = QMS_ACTIVE;
  295.   ObtainSemaphore (&qn -> qn_Semaphore);
  296.   AddTail (&qn -> qn_List, (Node *) msg);
  297.   qh -> qh_un.qhs.qhs_MsgCount ++;
  298.   if (qn -> qn_Read)
  299.   {
  300.     for (qh = (QueueHandle *) qn -> qn_Handles.lh_Head; qh -> qh_MinNode.mln_Succ;
  301.          qh = (QueueHandle *) qh -> qh_MinNode.mln_Succ)
  302.     {
  303.       if (qh -> qh_Mode == QMODE_LISTEN)
  304.         Signal (qh -> qh_SigTask, qh -> qh_SigMask);
  305.     }
  306.   }
  307.   else
  308.     RemoveAndReply (msg);
  309.   ReleaseSemaphore (&qn -> qn_Semaphore);
  310. }
  311.  
  312. ADDTABL_2(LIBQRemMsg,a0,a1);
  313.  
  314. void LIBRT
  315. LIBQRemMsg (REG(a0) QHandle qhandle, REG(a1) QMessage *msg)
  316. {
  317.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  318.  
  319.   ObtainSemaphore (&qn -> qn_Semaphore);
  320.   RemoveAndReply (msg);
  321.   ReleaseSemaphore (&qn -> qn_Semaphore);
  322. }
  323.  
  324. ADDTABL_1(LIBQGetMsg,a0);
  325.  
  326. QMessage * LIBRT
  327. LIBQGetMsg (REG(a0) QHandle qhandle)
  328. {
  329.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  330.   QueueHandle *qh = (QueueHandle *) qhandle;
  331.   QMessage *msg, *next;
  332.  
  333.   ObtainSemaphore (&qn -> qn_Semaphore);
  334.  
  335.   /* A server ( QMODE_SEND ) retrieves a replied message */
  336.  
  337.   if (qh -> qh_Mode == QMODE_SEND)
  338.   {
  339.     if (msg = (QMessage *) RemTail (&qh -> qh_un.qhs.qhs_ReplyList))
  340.       qh -> qh_un.qhs.qhs_MsgCount --;
  341.     ReleaseSemaphore (&qn -> qn_Semaphore);
  342.     return msg;
  343.   }
  344.  
  345.   /* A client ( QMODE_LISTEN ) reads a message */
  346.  
  347.   if (msg = qh -> qh_un.qhl.qhl_Message)
  348.   {
  349.     /* automagically reply the current message */
  350.  
  351.     next = (QMessage *) msg -> qm_MinNode.mln_Succ;
  352.     ReplyQMessage (qn, msg);
  353.     msg = next;
  354.   }
  355.   else /* no current message, start at marker or list head */
  356.   {
  357.     if (msg = (QMessage *) qh -> qh_un.qhl.qhl_MinNode.mln_Succ)
  358.     {
  359.       ClearMarker (qh);
  360.     }
  361.     else
  362.       msg = (QMessage *) qn -> qn_List.lh_Head;
  363.   }
  364.   for (;;) /* find the next message */
  365.   {
  366.     if (!(next = (QMessage *) msg -> qm_MinNode.mln_Succ))
  367.     {
  368.       SetMarker (qh, (QMessage *) msg -> qm_MinNode.mln_Pred);
  369.       msg = NULL; /* no message available */
  370.       break;
  371.     }
  372.     if (msg -> qm_Status & QMS_INACTIVE)
  373.     {
  374.       if (msg -> qm_Status == QMS_REMOVED)
  375.         RemoveAndReply (msg);
  376.       msg = next;
  377.       continue;
  378.     }
  379.     msg -> qm_Refs ++;
  380.     break;
  381.   }
  382.   qh -> qh_un.qhl.qhl_Message = msg;
  383.   ReleaseSemaphore (&qn -> qn_Semaphore);
  384.   return msg;
  385. }
  386.  
  387. ADDTABL_1(LIBQReplyMsg,a0);
  388.  
  389. ULONG LIBRT
  390. LIBQReplyMsg (REG(a0) QHandle qhandle)
  391. {
  392.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  393.   QueueHandle *qh = (QueueHandle *) qhandle;
  394.   QMessage *msg, *next;
  395.  
  396.   if (!(msg = qh -> qh_un.qhl.qhl_Message))
  397.     return 0;
  398.  
  399.   ObtainSemaphore (&qn -> qn_Semaphore);
  400.  
  401.   SetMarker (qh, msg);
  402.  
  403.   next = (QMessage *) msg -> qm_MinNode.mln_Succ;
  404.   ReplyQMessage (qn, msg);
  405.  
  406.   if (next)
  407.   {
  408.     for (msg = next; next = (QMessage *) msg -> qm_MinNode.mln_Succ; msg = next)
  409.     {
  410.       if (msg -> qm_Status == QMS_ACTIVE)
  411.       {
  412.     /* Since QReplyMsg() means you're not going to read more messages now
  413.        it's probably a good idea to remind you that there's more.
  414.     */
  415.     Signal (qh -> qh_SigTask, qh -> qh_SigMask);
  416.     break;
  417.       }
  418.     }
  419.   }
  420.   qh -> qh_un.qhl.qhl_Message = NULL;
  421.   ReleaseSemaphore (&qn -> qn_Semaphore);
  422.   return 1;
  423. }
  424.  
  425. ADDTABL_1(LIBQFlush,a0);
  426.  
  427. ULONG LIBRT
  428. LIBQFlush (REG(a0) QHandle qhandle)
  429. {
  430.   QueueNode *qn = ((QueueHandle *) qhandle) -> qh_QNode;
  431.   QMessage *msg, *next;
  432.  
  433.   ObtainSemaphore (&qn -> qn_Semaphore);
  434.   msg = (QMessage *) qn -> qn_List.lh_Head;
  435.   while (next = (QMessage *) msg -> qm_MinNode.mln_Succ)
  436.   {
  437.     if (msg -> qm_Status != QMS_MARKER)
  438.       RemoveAndReply (msg);
  439.     msg = next;
  440.   }
  441.   ReleaseSemaphore (&qn -> qn_Semaphore);
  442.   return 1;
  443. }
  444.  
  445. /* new in 3.0 */
  446.  
  447. ADDTABL_1(LIBQAllocMsg,d0);
  448.  
  449. QMessage * LIBRT
  450. LIBQAllocMsg (REG(d0) ULONG size)
  451. {
  452.   QMessage *msg;
  453.  
  454.   if (!(msg = AllocMem (sizeof (QMessage), MEMF_PUBLIC | MEMF_CLEAR)))
  455.     return NULL;
  456.   if (size)
  457.   {
  458.     if (!(msg -> qm_Data = AllocMem (size, MEMF_SHARED_READ)))
  459.     {
  460.       FreeMem (msg, sizeof (QMessage));
  461.       return NULL;
  462.     }
  463.   }
  464.   return msg;
  465. }
  466.  
  467. ADDTABL_2(LIBQFreeMsg,a0,d0);
  468.  
  469. void LIBRT
  470. LIBQFreeMsg (REG(a0) QMessage *msg, REG(d0) ULONG size)
  471. {
  472.   if (msg -> qm_Data)
  473.     FreeMem (msg -> qm_Data, size);
  474.   FreeMem (msg, sizeof (QMessage));
  475. }
  476.  
  477. ADDTABL_END();
  478.